/*
 * tts engine based on alibaba cloud tts
 *
 * easycallcenter365@126.com
 *
 * mod_aliyun_tts.cpp -- Alibaba Cloud TTS module for freeswitch
 *
 * compile:

 g++ -shared -o mod_aliyun_tts.so -fPIC -g -O -ggdb -std=c++11 -Wall  \
 -I../../../../libs/libhv-master/master/libhv/include/hv/  \
 -I../../../../libs/cpputils/ -I../../../../src/include/  \
 -I../../../../libs/libteletone/src/   mod_aliyun_tts.cpp   \
 -L/usr/local/freeswitchvideo/lib/  -lfreeswitch  \
 -L/usr/local/lib/  -lhv    -lcurl  -lpthread   -lssl -lcrypto

 debug:
 g++ -shared -o mod_aliyun_tts.so -fPIC -fsanitize=address -g -O0 -ggdb -std=c++11 -Wall  \
 -I../../../../libs/libhv-master/master/libhv/include/hv/  \
 -I../../../../libs/cpputils/ -I../../../../src/include/  \
 -I../../../../libs/libteletone/src/   mod_aliyun_tts.cpp   \
 -L/usr/local/freeswitchvideo/lib/  -lfreeswitch  \
 -L/usr/local/lib/  -lhv    -lcurl  -lpthread   -lssl -lcrypto -fsanitize=address



usage:
<action application="speak" data="<engine>|<voice>|<text>|[timer_name]"/>

Example:

first request:
<action application="speak" data="aliyun_tts|aixia|FreeSWITCH is awesome"/>

resume session based on the last request:
<action application="aliyun_tts_resume" data="FreeSWITCH is powerful!"/>

once speak completed, the session will be closed automatically.
you need to use speak to reopen the session again if you want to speak again.


 */

#include "WebSocketClient.h" 
#include "hpp/AliyunToken.hpp"
#include "hpp/UriUtils.hpp"
#include "hpp/sha_256.hpp"
#include "rapidjson/document.h"
#include "rapidjson/stringbuffer.h"
#include "rapidjson/writer.h"
#include <EventLoopThreadPool.h>
#include <atomic>
#include <condition_variable>
#include <cstdio>
#include <curl/curl.h>
#include <iostream>
#include <memory>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <switch.h>

using namespace hv;
using namespace std;
using namespace rapidjson;

#define THREAD_POOL_SIZE 20
#define aliyun_tts_mod "aliyun_tts_mod"

SWITCH_MODULE_LOAD_FUNCTION(mod_aliyun_tts_load);
SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_aliyun_tts_shutdown);
extern "C" {
SWITCH_MODULE_DEFINITION(mod_aliyun_tts, mod_aliyun_tts_load, mod_aliyun_tts_shutdown, NULL);
}

static struct {
	char *server_url;
	int ws_conn_timeout_ms;
	int sample_rate;
	char *access_key_id;
	char *access_key_secret;
	char *app_key;
	char *voice_name;
	int voice_volume;
	int speech_rate;
	int write_pcm_enable;
	int buffer_size;
	int buffer_max_size;
	string token;
	string stop_cmd = "<StopSynthesis/>";
	long token_expire_time;
	switch_mutex_t *MUTEX;
	switch_memory_pool_t *pool;
} globals;

typedef struct aliyuntts {
	char *voice;
	char *task_uuid;
	char *uuid;
	char *app_key;
	FILE *pcm_file;
	std::shared_ptr<WebSocketClient> ws;
	std::atomic<int> ws_connected;
	std::atomic<int> tts_stream_text_end;
	std::atomic<int> ws_closed;
	std::atomic<int> synthesisCompleted;
	std::atomic<int> recv_first_data;
	std::atomic<int> send_request_num;
	std::atomic<int> recv_completed_num;
	std::atomic<int> tts_closed;
	std::atomic<int> conn_finished;
	std::atomic<int> read_audio_data;
	std::atomic<int> play_back_stopped;

	std::mutex *mtx;
	std::condition_variable *cv;
	std::mutex *mtx_for_close;
	std::condition_variable *cv_for_close;

	switch_thread_cond_t *cond;
	switch_mutex_t *cond_mutex; 
	switch_mutex_t *buffer_mutex; 

	switch_memory_pool_t *pool;
	switch_buffer_t *audio_buffer;
	switch_core_session_t *session;

	// tts-text queue
	switch_queue_t *tts_text_queue;

} aliyuntts_t;

static void destroy(aliyuntts_t **pvtArg)
{
	if (!pvtArg || !*pvtArg) return;

	aliyuntts_t *pvt = *pvtArg;
	 
	if (!pvt->tts_closed.load()) {
		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "wait for tts_closed , uuid=%s\n", pvt->uuid);
		switch_mutex_lock(pvt->cond_mutex);
		switch_thread_cond_wait(pvt->cond, pvt->cond_mutex);
		switch_mutex_unlock(pvt->cond_mutex);
	}
	if (!pvt->ws_closed.load()) { 
		std::this_thread::sleep_for(std::chrono::milliseconds(1000));
		if (!pvt->ws_closed.load()) {
			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_WARNING, "wait for tts websocket closed , uuid=%s\n", pvt->uuid);
		}
	}
	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "tts is closed, uuid=%s\n", pvt->uuid);
	 
	if (pvt->mtx) {
		delete pvt->mtx;
		pvt->mtx = nullptr;
	}
	if (pvt->cv) {
		delete pvt->cv;
		pvt->cv = nullptr;
	}
	if (pvt->mtx_for_close) {
		delete pvt->mtx_for_close;
		pvt->mtx_for_close = nullptr;
	}
	if (pvt->cv_for_close) {
		delete pvt->cv_for_close;
		pvt->cv_for_close = nullptr;
	}
	if (pvt->ws) { pvt->ws = NULL; } // this will trigger the destructor of WebSocketClient

	if (pvt->audio_buffer) {
		switch_mutex_lock(pvt->buffer_mutex);
		switch_buffer_zero(pvt->audio_buffer);
		switch_mutex_unlock(pvt->buffer_mutex);
		switch_buffer_destroy(&pvt->audio_buffer);
		pvt->audio_buffer = NULL;
	}

	if (pvt->pcm_file) { fclose(pvt->pcm_file); }

	switch_mutex_destroy(pvt->cond_mutex);
	switch_mutex_destroy(pvt->buffer_mutex);
	switch_thread_cond_destroy(pvt->cond);

	if (pvt->tts_text_queue) {
		void *pop;
		
		while (switch_queue_trypop(pvt->tts_text_queue, &pop) == SWITCH_STATUS_SUCCESS) {}

		switch_queue_term(pvt->tts_text_queue);
	}
	pvt->tts_text_queue = NULL;

	if (pvt->pool) {
		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,
						  "switch_core_destroy_memory_pool, pvt->pool memory address=%p, uuid=%s\n",
						  (void *)(pvt->pool), pvt->uuid);
		switch_core_destroy_memory_pool(&pvt->pool);
	}
	pvt->uuid = NULL;
	pvt->pool = NULL;
	pvt->cond_mutex = NULL;
	pvt->cond = NULL;
	pvt = NULL;
};

/// <summary>
/// 获取token
/// </summary>
/// <param name="accessKeyId"></param>
/// <param name="accessKeySecret"></param>
/// <returns>成功返回token字符串，失败返回空字符串</returns>
string getToken(std::string &accessKeyId, std::string &accessKeySecret, aliyuntts_t *pvt)
{
	long now = switch_time_now() / 1000 / 1000;
	if (globals.token_expire_time - now > 3600) { return globals.token; }

	switch_mutex_lock(globals.MUTEX);

	if (globals.token_expire_time - now < 3600) {
		if (globals.token_expire_time > 0L) {
			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,
							  "Try to create token. Token is going to be expired, now:%ld, expire:%ld, pvt->uuid=%s.\n",
							  now, globals.token_expire_time, pvt->uuid);
		}
		AccessToken accessToken(accessKeyId, accessKeySecret);
		try {
			accessToken.apply();
			long expireTime = accessToken.getExpireTime();
			if (expireTime > 0) {
				globals.token = accessToken.getToken();
				globals.token_expire_time = expireTime;
				switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,
								  " getToken successfully, token=%s, expireTime=%ld, pvt->uuid=%s.\n",
								  globals.token.c_str(), expireTime, pvt->uuid);
			}
		} catch (const std::exception &e) {
			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, " getToken error, msg = %s, pvt->uuid=%s. \n",
							  e.what(), pvt->uuid);
		}
	}

	switch_mutex_unlock(globals.MUTEX);

	return globals.token;
}

string genUUID()
{
	char uuid_str[SWITCH_UUID_FORMATTED_LENGTH + 1];
	switch_uuid_str(uuid_str, sizeof(uuid_str));

	string uuid_str_tmp(uuid_str);
	uuid_str_tmp.erase(std::remove(uuid_str_tmp.begin(), uuid_str_tmp.end(), '-'), uuid_str_tmp.end());
	return uuid_str_tmp;
}

void customDeleter(WebSocketClient *wsClient)
{
	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "Custom deleter called for WebSocketClient.\n");
	delete wsClient;
};

  
std::string prepare_tts_request(const char *text, aliyuntts_t *pvt)
{
	Document document;
	document.SetObject();
	string msg_id_str = genUUID();

	rapidjson::Document::AllocatorType &allocator = document.GetAllocator();
	rapidjson::Value header(rapidjson::kObjectType);
	header.AddMember("namespace", rapidjson::StringRef("SpeechSynthesizer"), allocator);
	header.AddMember("name", rapidjson::StringRef("StartSynthesis"), allocator);
	header.AddMember("message_id", rapidjson::StringRef(msg_id_str.c_str()), allocator);	 
	header.AddMember("appkey", rapidjson::StringRef(pvt->app_key), allocator);	 
	header.AddMember("task_id", rapidjson::StringRef(pvt->task_uuid), allocator);
	document.AddMember("header", header, allocator);

	rapidjson::Value payload(rapidjson::kObjectType);
	payload.AddMember("text", rapidjson::StringRef(text), allocator);
	payload.AddMember("voice", rapidjson::StringRef(pvt->voice), allocator);
	payload.AddMember("format", rapidjson::StringRef("wav"), allocator);
	payload.AddMember("sample_rate", globals.sample_rate, allocator);
	payload.AddMember("volume", globals.voice_volume, allocator);
	payload.AddMember("speech_rate", globals.speech_rate, allocator);
	payload.AddMember("pitch_rate", 0, allocator);
	payload.AddMember("enable_subtitle", false, allocator);
	document.AddMember("payload", payload, allocator);

	rapidjson::StringBuffer buffer;
	rapidjson::Writer<rapidjson::StringBuffer> writer(buffer);
	document.Accept(writer);

	string json = buffer.GetString();
	return json;
};

/// <summary>
/// connect to  tts server
/// </summary>
/// <returns>if connected return 1, otherwise return 0</returns>
static int connect_ws(aliyuntts_t *pvt)
{
	string key = string(globals.access_key_id);
	string secret = string(globals.access_key_secret);

	// Attempt to fetch the token from the channel as the first option.
	switch_channel_t *channel = switch_core_session_get_channel(pvt->session);
	const char *token_var = switch_channel_get_variable(channel, "aliyun_tts_token");

	string token = "";

	if (token_var) {
		token = string(token_var);
		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "%s, Successfully fetched the token from the channel. \n", pvt->uuid);
	} else {
		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_WARNING,
						  "%s, Retrieve the token from the channel failed! \n", pvt->uuid);
		token = getToken(key, secret, pvt);
	}

	if (token.length() == 0) { return 0; }

	string url = string(globals.server_url) + string("?token=") + token;
	string task_uuid_str = genUUID();
	pvt->task_uuid = switch_core_strdup(pvt->pool, task_uuid_str.c_str());

	if (globals.write_pcm_enable) {
		char pcm_path[100] = {0};
		sprintf(pcm_path, "/tmp/aliyun_tts_pcm_file_%s", task_uuid_str.c_str());
		pvt->pcm_file = fopen(pcm_path, "wb");
		if (!pvt->pcm_file) {
			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "FAILED to open file=%s \n", pcm_path);
		} else {
			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "open file=%s successfully \n", pcm_path);
		}
	}

	std::shared_ptr<WebSocketClient> ws_client(new WebSocketClient(), customDeleter);
	pvt->ws = ws_client;
	pvt->ws->setConnectTimeout(globals.ws_conn_timeout_ms); 

	pvt->ws->onopen = [pvt]() {
		pvt->ws_connected.store(1);
		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "TTS Server is ready,  pvt->uuid=%s \n", pvt->uuid);

		if (!pvt->conn_finished.load()) {
			std::unique_lock<std::mutex> lock(*pvt->mtx);
			pvt->conn_finished.store(1);
			pvt->cv->notify_one();
		}
	};

	pvt->ws->onclose = [pvt]() {
		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "recv TTS Server disconnected msg, pvt->uuid=%s \n",
						  pvt->uuid);
		if (!pvt->conn_finished.load()) {
			std::unique_lock<std::mutex> lock(*pvt->mtx);
			pvt->conn_finished.store(1);
			pvt->cv->notify_one();
		}

		std::unique_lock<std::mutex> lock(*pvt->mtx_for_close);
		pvt->ws_closed.store(1);
		// send signal to main thread
		pvt->cv_for_close->notify_one();
	};

	pvt->ws->onmessage = [pvt](int opcode, const std::string &msg) {
		if (opcode == 1) {
			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "recv TTS Server text data: %s, pvt->uuid=%s \n",
							  msg.c_str(), pvt->uuid);

			Document document;
			document.Parse(msg.c_str());

			if (!document.IsObject()) { return; }
			if (document.HasParseError()) {
				switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "parse json error, code = %d, pvt->uuid=%s \n",
								  document.GetParseError(), pvt->uuid);
				return;
			}

			// 检查是否包含 "header" 字段
			if (document.HasMember("header") && document["header"].IsObject()) {
				const rapidjson::Value &header = document["header"];

				// 检查是否包含 "status" 字段，如果有则打印状态
				if (header.HasMember("status") && header["status"].IsInt()) {
					int status = header["status"].GetInt();
					// 判断是否为成功响应
					if (status == 20000000) {
						switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,
										  "TTS Server response status code  20000000, pvt->uuid=%s.\n", pvt->uuid);
					}
				}

				if (header.HasMember("name") && header["name"].IsString()) {
					std::string name = header["name"].GetString();
					if (name == "SynthesisCompleted") { 						
						pvt->recv_completed_num.fetch_add(1);						 
						pvt->synthesisCompleted.store(1); 
						pvt->recv_first_data.store(0);						 

						switch_mutex_lock(pvt->cond_mutex);
						switch_thread_cond_signal(pvt->cond);
						switch_mutex_unlock(pvt->cond_mutex);						
					}
				}
			}

		} else if (opcode == 2) {
			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG,
							  "recv TTS Server binary data, size = %ld, pvt->uuid=%s. \n", msg.size(), pvt->uuid);

			bool is_first_data = pvt->recv_first_data.load() == 0;
			if (is_first_data) { 
				pvt->recv_first_data.store(1);
			} else {
				if (msg.size() > 0) {
					switch_mutex_lock(pvt->buffer_mutex);

					size_t free_space = switch_buffer_freespace(pvt->audio_buffer);
					if(msg.size() <= free_space) {
						switch_buffer_write(pvt->audio_buffer, msg.c_str(), msg.size());
					} else {
						switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_WARNING, "Buffer full! Dropping %zu bytes, pvt->uuid=%s.\n", msg.size(), pvt->uuid);
					}

					switch_mutex_unlock(pvt->buffer_mutex);
				}
			}
		 

			if (!pvt->read_audio_data.load()) {
				std::unique_lock<std::mutex> lock(*pvt->mtx);
				pvt->read_audio_data.store(1);
				pvt->cv->notify_one();
			}

			if (pvt->play_back_stopped.load()) {
				pvt->play_back_stopped.store(0);
				switch_event_t *event = NULL;
				switch_channel_t *channel = switch_core_session_get_channel(pvt->session);
				if (switch_event_create(&event, SWITCH_EVENT_PLAYBACK_START) == SWITCH_STATUS_SUCCESS) {
					switch_channel_event_set_data(channel, event);
					switch_event_fire(&event);
					switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,
									  "fire PLAYBACK_START event, pvt->uuid=%s. \n", pvt->uuid);
				}
			}

			if (pvt->pcm_file) { fwrite(msg.c_str(), sizeof(char), msg.size(), pvt->pcm_file); }
		}
	};

	reconn_setting_t reconn;
	reconn.max_retry_cnt = 0; // disable reconnect
	pvt->ws->connect_timeout = globals.ws_conn_timeout_ms;
	chrono::steady_clock::time_point start = chrono::steady_clock::now();
	chrono::steady_clock::time_point end;
	pvt->ws->setReconnect(&reconn);

	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,
					  "Try to connect aliyun tts server=%s , task_uuid=%s, pvt->uuid=%s \n", url.c_str(),
					  task_uuid_str.c_str(), pvt->uuid);
	pvt->ws->open(url.c_str());

	// wait for connection finished
	{
		std::unique_lock<std::mutex> lock(*pvt->mtx);
		pvt->cv->wait_for(lock, std::chrono::milliseconds(pvt->ws->connect_timeout),
						  [pvt] { return pvt->conn_finished.load(); });
	}

	end = chrono::steady_clock::now();
	auto duration = chrono::duration_cast<chrono::microseconds>(end - start).count();
	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,
					  "tts-server connect cost time = %Lf microseconds,pvt->uuid=%s \n",
					  duration < 0.0L ? 0.0L : duration, pvt->uuid);

	if (!pvt->ws_connected.load()) {
		pvt->ws->close();
		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "TTS Server connected failed. pvt->uuid=%s\n",
						  pvt->uuid);
		{
			std::unique_lock<std::mutex> lock(*pvt->mtx_for_close);
			pvt->cv_for_close->wait_for(lock, std::chrono::milliseconds(3500), [pvt] { return pvt->ws_closed.load(); });
			pvt->ws->onclose = NULL;
			if (!pvt->ws_closed.load()) { std::this_thread::sleep_for(std::chrono::milliseconds(1000)); }
		}
	}

	return pvt->ws_connected.load();
}

static void send_tts_request(aliyuntts_t *pvt, char *tts_text)
{
	if (!pvt || !tts_text) { return; }
   
	if (pvt->ws_closed.load() || !pvt->ws_connected.load() || pvt->tts_closed.load()) { return; }

	if (pvt->ws) {
		string msg_id_str = genUUID();

		std::string tts_str(tts_text);
		if (tts_str == globals.stop_cmd) { 
			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "recv StopSynthesis cmd, pvt->uuid=%s. \n",
							  pvt->uuid); 
			pvt->tts_stream_text_end.store(1); 
		} else {

            // Prepare TTS request data
			std::string request_data = prepare_tts_request(tts_text, pvt);
			if (request_data.empty()) {
				return;
			} 
			pvt->synthesisCompleted.store(0);

			pvt->ws->send(request_data);
			pvt->send_request_num.fetch_add(1);
			switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "TTS request has been sent, text length=%s, pvt->uuid=%s \n",
							   tts_text, pvt->uuid);
		}
	}
}

static switch_status_t aliyuntts_speech_open(switch_speech_handle_t *sh, const char *voice_name, int rate, int channels,
											 switch_speech_flag_t *flags)
{
	switch_memory_pool_t *pool;
	if (switch_core_new_memory_pool(&pool) != SWITCH_STATUS_SUCCESS) { return SWITCH_STATUS_FALSE; }

	aliyuntts_t *pvt = (aliyuntts_t *)switch_core_alloc(pool, sizeof(aliyuntts_t));
	sh->native_rate = globals.sample_rate;

	pvt->pool = pool;

	// if the voice name is not set, use the default voice name from config file.
	if (voice_name) {
		pvt->voice = switch_core_strdup(pool, voice_name);
	} else {
		pvt->voice = switch_core_strdup(pool, globals.voice_name);
	}

	pvt->session = sh->session;
	pvt->uuid = switch_core_strdup(pool, switch_core_session_get_uuid(sh->session));
	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,
					  "on aliyun_tts_speech_open event...pvt->uuid=%s, pvt->pool=%p\n", pvt->uuid, (void *)pool);

	switch_channel_t *channel = switch_core_session_get_channel(pvt->session);
	const char *app_key_var = switch_channel_get_variable(channel, "aliyun_tts_app_key");
	if (app_key_var) {
		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE,
						  "%s, Successfully fetched the app_key from the channel. app_key=%s \n", pvt->uuid,
						  app_key_var);
		pvt->app_key = switch_core_strdup(pool, (char *)app_key_var);
	} else {
		pvt->app_key = globals.app_key;
	}

	// record latest request time
	pvt->ws_connected.store(0);
	pvt->ws_closed.store(0);
	pvt->tts_closed.store(0);
	pvt->synthesisCompleted.store(0);
	pvt->recv_first_data.store(0);
	pvt->conn_finished.store(0);
	pvt->read_audio_data.store(0);
	pvt->play_back_stopped.store(1);
	pvt->tts_stream_text_end.store(0);
	pvt->send_request_num.store(0);
	pvt->recv_completed_num.store(0);

	pvt->mtx = new std::mutex();
	pvt->cv = new std::condition_variable();
	pvt->mtx_for_close = new std::mutex();
	pvt->cv_for_close = new std::condition_variable();

	if (switch_mutex_init(&pvt->cond_mutex, SWITCH_MUTEX_NESTED, pvt->pool) != SWITCH_STATUS_SUCCESS) {
		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "unable to create pvt->cond_mutex, uuid=%s \n", pvt->uuid);
		pvt->tts_closed.store(1);
		destroy(&pvt);
		return SWITCH_STATUS_FALSE;
	}

	if (switch_mutex_init(&pvt->buffer_mutex, SWITCH_MUTEX_NESTED, pvt->pool) != SWITCH_STATUS_SUCCESS) {
		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Failed to init buffer mutex\n");
		pvt->tts_closed.store(1);
		destroy(&pvt);
		return SWITCH_STATUS_FALSE;
	}


	if (switch_thread_cond_create(&pvt->cond, pvt->pool) != SWITCH_STATUS_SUCCESS) {
		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "unable to create pvt condition variable");
		pvt->tts_closed.store(1);
		destroy(&pvt);
		return SWITCH_STATUS_FALSE;
	}	

	switch_buffer_create_dynamic(&pvt->audio_buffer, globals.buffer_size, globals.buffer_size, globals.buffer_max_size);
	
	switch_queue_create(&pvt->tts_text_queue, 100, pool);

	switch_channel_set_private(channel, aliyun_tts_mod, pvt);

	if (!connect_ws(pvt)) {
		pvt->tts_closed.store(1);
		destroy(&pvt);
		return SWITCH_STATUS_FALSE;
	}

	// start new thread to handle pop tts-text queue
	std::thread([pvt] {
		long thread_start_time = switch_time_now() / 1000;
		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,
						  "TTS processing thread started, timestamp=%ld, pvt->uuid=%s \n", thread_start_time,
						  pvt->uuid);
		int first_run = 1;

		while (!pvt->tts_closed.load() && !pvt->ws_closed.load()) {
			void *pop;
			while (switch_queue_trypop(pvt->tts_text_queue, &pop) == SWITCH_STATUS_SUCCESS && pop &&
				   !pvt->tts_closed.load()) {
				char *tts_text = (char *)pop;
				switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,
								  "recv tts-text from queue, tts-text=%s, pvt->uuid=%s, try to feed tts. \n", tts_text,
								  pvt->uuid);
				if (first_run) {
					first_run = 0;
					send_tts_request(pvt, tts_text); 
				}else{				
					if (pvt->synthesisCompleted.load()){
					   send_tts_request(pvt, tts_text); 
					} else { 
						switch_log_printf(
							SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,
							"Wait for synthesisCompleted event, tts-text=%s, pvt->uuid=%s, feed tts. \n",
							tts_text, pvt->uuid);

						switch_mutex_lock(pvt->cond_mutex);
						switch_thread_cond_timedwait(pvt->cond, pvt->cond_mutex, 7 * 1000 * 1000);
						switch_mutex_unlock(pvt->cond_mutex);
						  
						if (!pvt->tts_closed.load() || !pvt->ws_closed.load()) {
							if (pvt->synthesisCompleted.load()) {
								send_tts_request(pvt, tts_text);
							} else {
								switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE,
												  "Wait-time-out for synthesisCompleted event, tts-text=%s, "
												  "pvt->uuid=%s, feed tts error. \n",
												  tts_text, pvt->uuid);
							}
						}
					}
				}
				std::this_thread::sleep_for(std::chrono::milliseconds(50));
			}
			std::this_thread::sleep_for(std::chrono::milliseconds(50));
		}

		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "tts-text pop thread is end, pvt->uuid=%s. \n",
						  pvt->uuid);

		{
			std::unique_lock<std::mutex> lock(*pvt->mtx_for_close);
			pvt->cv_for_close->wait_for(lock, std::chrono::milliseconds(3500), [pvt] { return pvt->ws_closed.load(); });
			pvt->ws->onclose = NULL;
			if (!pvt->ws_closed.load()) { std::this_thread::sleep_for(std::chrono::milliseconds(1000)); }
		}
		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,
						  "Close tts server ws connection is done, pvt->uuid=%s. \n", pvt->uuid);

		//g_pvt_manager.register_object(pvt);
		aliyuntts_t *pvtArg = (aliyuntts_t *)pvt;
		destroy(&pvtArg);
	}).detach();

	return SWITCH_STATUS_SUCCESS;
}

static switch_status_t aliyuntts_speech_close(switch_speech_handle_t *sh, switch_speech_flag_t *flags)
{
	aliyuntts_t *pvt;
	switch_channel_t *channel = switch_core_session_get_channel(sh->session);
	pvt = (aliyuntts_t *)switch_channel_get_private(channel, aliyun_tts_mod);

	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "on aliyun_tts_speech_close event...pvt->uuid=%s. \n",
					  pvt->uuid);

	if (pvt->ws && !pvt->ws_closed.load()) { pvt->ws->close(); }

	pvt->tts_closed.store(1);
	switch_mutex_lock(pvt->cond_mutex);
	switch_thread_cond_signal(pvt->cond);
	switch_mutex_unlock(pvt->cond_mutex);
	 

	switch_event_t *event = NULL;
	if (switch_event_create(&event, SWITCH_EVENT_CUSTOM) == SWITCH_STATUS_SUCCESS) {
		event->subclass_name = strdup("TtsEvent");
		switch_channel_event_set_data(channel, event);
		switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "Tts-Event-Detail", "Speech-Closed");
		switch_event_fire(&event);
	}

	return SWITCH_STATUS_SUCCESS;
}

static switch_status_t aliyuntts_speech_feed_tts(switch_speech_handle_t *sh, char *text, switch_speech_flag_t *flags)
{
	aliyuntts_t *pvt;
	switch_channel_t *channel = switch_core_session_get_channel(sh->session);
	pvt = (aliyuntts_t *)switch_channel_get_private(channel, aliyun_tts_mod);

	if (!pvt || !pvt->pool || pvt->tts_closed.load() || pvt->ws_closed.load()) { return SWITCH_STATUS_FALSE; }
	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "on aliyun_tts_speech_feed_tts event...pvt->uuid=%s \n",
					  pvt->uuid);
	// add to tts text queue
	char *tts_text = switch_core_strdup(pvt->pool, text);
	switch_queue_push(pvt->tts_text_queue, tts_text);
	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "recv feed tts text=%s, ", tts_text);

	return SWITCH_STATUS_SUCCESS;
};

static void aliyuntts_speech_flush_tts(switch_speech_handle_t *sh)
{
	aliyuntts_t *pvt;
	switch_channel_t *channel = switch_core_session_get_channel(sh->session);
	pvt = (aliyuntts_t *)switch_channel_get_private(channel, aliyun_tts_mod);
	switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "on aliyun_tts_speech_flush_tts event...pvt->uuid=%s \n",
					  pvt->uuid);

	if (pvt->audio_buffer) {
		switch_mutex_lock(pvt->buffer_mutex);
		switch_buffer_zero(pvt->audio_buffer);
		switch_mutex_unlock(pvt->buffer_mutex);	
	}
}

static switch_status_t aliyuntts_speech_read_tts(switch_speech_handle_t *sh, void *data, size_t *datalen,
												 switch_speech_flag_t *flags)
{
	size_t bytes_read;
	aliyuntts_t *pvt;
	switch_channel_t *channel = switch_core_session_get_channel(sh->session);
	pvt = (aliyuntts_t *)switch_channel_get_private(channel, aliyun_tts_mod);

	if (!pvt || !pvt->audio_buffer) {
		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Failed to get private data!\n");
		return SWITCH_STATUS_FALSE;
	}

	switch_mutex_lock(pvt->buffer_mutex);
	if ((bytes_read = switch_buffer_read(pvt->audio_buffer, data, *datalen))) { *datalen = bytes_read; }
	switch_mutex_unlock(pvt->buffer_mutex);

	if (!bytes_read) {

		if (pvt->read_audio_data.load() == 0 ) {
			{
				std::unique_lock<std::mutex> lock(*pvt->mtx);
				pvt->cv->wait_for(lock, std::chrono::milliseconds(5000),
								  [&pvt] { return pvt->read_audio_data.load() > 0; });
			}
		}

		// Determine maximum wait time
		const int max_wait_ms = 5000;
		const int wait_step_ms = 20; // Shorten each wait time, improve responsiveness
		int waited_ms = 0;

		while (waited_ms < max_wait_ms) {
			waited_ms += wait_step_ms;

			std::this_thread::sleep_for(std::chrono::milliseconds(wait_step_ms));

			switch_mutex_lock(pvt->buffer_mutex);
			bytes_read = switch_buffer_read(pvt->audio_buffer, data, *datalen);
			switch_mutex_unlock(pvt->buffer_mutex);

			if (bytes_read) {
				*datalen = bytes_read;
				break;
			}

			if (pvt->tts_stream_text_end.load() >0 && pvt->send_request_num.load() == pvt->recv_completed_num.load()) { 
				switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "recv tts_stream_text_end flag, and send_request_num == recv_completed_num\n");
				break;
			} 			
		}
	}
	 

	if (!bytes_read) {

		if (pvt->read_audio_data.load() > 0 && pvt->play_back_stopped.load() <= 0) {

			pvt->play_back_stopped.store(1);
			switch_event_t *event = NULL;
			if (switch_event_create(&event, SWITCH_EVENT_PLAYBACK_STOP) == SWITCH_STATUS_SUCCESS) {
				switch_channel_event_set_data(channel, event);
				switch_event_fire(&event);
				switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "fire PLAYBACK_STOP event, pvt->uuid=%s. \n",
								  pvt->uuid);
			}
		}
		 
		switch_log_printf(
			SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,
			"TTS read 0 bytes, tts_stream_text_end=%d, pvt->send_request_num=%d, pvt->recv_completed_num=%d \n",
			pvt->tts_stream_text_end.load(), pvt->send_request_num.load(), pvt->recv_completed_num.load());
	}

	return (bytes_read > 0) ? SWITCH_STATUS_SUCCESS : SWITCH_STATUS_FALSE;
}

static void aliyuntts_text_param_tts(switch_speech_handle_t *sh, char *param, const char *val) {}

static void aliyuntts_numeric_param_tts(switch_speech_handle_t *sh, char *param, int val) {}

static void aliyuntts_float_param_tts(switch_speech_handle_t *sh, char *param, double val) {}

/* internal function to load the aliyuntts configuration*/
static switch_status_t aliyuntts_load_config(void)
{
	char *cf = (char *)"aliyun_tts.conf";
	switch_xml_t cfg, xml = NULL, param, settings;
	switch_status_t status = SWITCH_STATUS_SUCCESS;

	// global default settings
	globals.server_url = (char *)"";
	globals.app_key = NULL;
	globals.voice_name = (char *)"";
	globals.buffer_size = 49152;
	globals.buffer_max_size = 4194304;
	globals.token_expire_time = 0;

	if (!(xml = switch_xml_open_cfg(cf, &cfg, NULL))) {
		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Open of %s failed\n", cf);
		status = SWITCH_STATUS_FALSE;
		goto done;
	}

	if ((settings = switch_xml_child(cfg, "settings"))) {
		for (param = switch_xml_child(settings, "param"); param; param = param->next) {
			char *var = (char *)switch_xml_attr_soft(param, "name");
			char *val = (char *)switch_xml_attr_soft(param, "value");
			if (strcasecmp(var, "server_url") == 0) {
				globals.server_url = switch_core_strdup(globals.pool, val);
			} else if (strcasecmp(var, "app_key") == 0) {
				globals.app_key = switch_core_strdup(globals.pool, val);
			} else if (strcasecmp(var, "voice_name") == 0) {
				globals.voice_name = switch_core_strdup(globals.pool, val);
			} else if (strcasecmp(var, "buffer_size") == 0) {
				globals.buffer_size = atoi(val);
			} else if (strcasecmp(var, "buffer_max_size") == 0) {
				globals.buffer_max_size = atoi(val);
			} else if (strcasecmp(var, "access_key_id") == 0) {
				globals.access_key_id = switch_core_strdup(globals.pool, val);
			} else if (strcasecmp(var, "access_key_secret") == 0) {
				globals.access_key_secret = switch_core_strdup(globals.pool, val);
			} else if (strcasecmp(var, "ws_conn_timeout_ms") == 0) {
				globals.ws_conn_timeout_ms = atoi(val);
			} else if (strcasecmp(var, "sample_rate") == 0) {
				globals.sample_rate = atoi(val);
			} else if (strcasecmp(var, "write_pcm_enable") == 0) {
				globals.write_pcm_enable = switch_true(val);
			} else if (strcasecmp(var, "voice_volume") == 0) {
				globals.voice_volume = switch_safe_atoi(val, 50);
			} else if (strcasecmp(var, "speech_rate") == 0) {
				globals.speech_rate = switch_safe_atoi(val, 0);
			}
		}
	}

	switch_log_printf(
		SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO,
		"aliyuntts_load_config, server_url=%s, app_key=%s, sample_rate=%d, voice_volume=%d, globals.speech_rate=%d. \n",
		globals.server_url, globals.app_key, globals.sample_rate, globals.voice_volume, globals.speech_rate);

done:
	if (xml) { switch_xml_free(xml); }

	return status;
}

static void do_config_load(void)
{
	switch_mutex_lock(globals.MUTEX);
	aliyuntts_load_config();
	switch_mutex_unlock(globals.MUTEX);
}

SWITCH_STANDARD_APP(aliyuntts_resume_session_function)
{
	aliyuntts_t *pvt;
	switch_channel_t *channel = switch_core_session_get_channel(session);
	pvt = (aliyuntts_t *)switch_channel_get_private(channel, aliyun_tts_mod);
	if (!pvt || !pvt->pool || pvt->tts_closed.load()) {
		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR,
						  "%s resume_tts command failed to get private data.\n", switch_core_session_get_uuid(session));
		return;
	}

	if (!pvt || !pvt->pool || pvt->tts_closed.load() || pvt->ws_closed.load()) { return; }
	if (!zstr(data)) {
		// Add to TTS text queue
		char *tts_text = switch_core_strdup(pvt->pool, data);
		switch_queue_push(pvt->tts_text_queue, tts_text);
		switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "recv resume tts text=%s, ", tts_text);

	} else {
		switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_WARNING,
						  "%s resume_tts command requires 1 parameter: TTS text.\n", pvt->uuid);
	}
};

SWITCH_MODULE_LOAD_FUNCTION(mod_aliyun_tts_load)
{
	switch_speech_interface_t *tts_interface;

	switch_mutex_init(&globals.MUTEX, SWITCH_MUTEX_NESTED, pool);
	globals.pool = pool;

	do_config_load();

	/* connect my internal structure to the blank pointer passed to me */
	switch_application_interface_t *app_interface;
	*module_interface = switch_loadable_module_create_module_interface(pool, modname);

	SWITCH_ADD_APP(app_interface, "aliyun_tts_resume", aliyun_tts_mod, aliyun_tts_mod,
				   aliyuntts_resume_session_function, "", SAF_NONE);

	tts_interface = (switch_speech_interface_t *)switch_loadable_module_create_interface(*module_interface,
																						 SWITCH_SPEECH_INTERFACE);
	tts_interface->interface_name = "aliyun_tts";
	tts_interface->speech_open = aliyuntts_speech_open;
	tts_interface->speech_close = aliyuntts_speech_close;
	tts_interface->speech_feed_tts = aliyuntts_speech_feed_tts;
	tts_interface->speech_read_tts = aliyuntts_speech_read_tts;
	tts_interface->speech_flush_tts = aliyuntts_speech_flush_tts;
	tts_interface->speech_text_param_tts = aliyuntts_text_param_tts;
	tts_interface->speech_numeric_param_tts = aliyuntts_numeric_param_tts;
	tts_interface->speech_float_param_tts = aliyuntts_float_param_tts;

	/* indicate that the module should continue to be loaded */
	return SWITCH_STATUS_SUCCESS;
}

SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_aliyun_tts_shutdown) { return SWITCH_STATUS_UNLOAD; }
